#include "mpHttpDispatcher.h"

#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>
#include <boost/asio/placeholders.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/basic_waitable_timer.hpp>

#if defined(MAPLE_MODE_DEBUGGING)
#include "mpLocaleUtils.h"
#endif

#include "mpHttpParams.h"
#include "mpHttpException.h"
#include "mpLog.h"

namespace MP{
namespace HTTP{
namespace INTERNAL{

//private member
namespace{

#if defined(MAPLE_MODE_DEBUGGING)

std::string endpointToString(const boost::asio::ip::tcp::endpoint& endpoint) {
    std::ostringstream ss;
    ss << endpoint;
    if (endpoint.protocol() == boost::asio::ip::tcp::v4()) {
        ss << ", v4";
    } else if (endpoint.protocol() == boost::asio::ip::tcp::v6()) {
        ss << ", v6";
    }
    return ss.str();
}

#endif

} //private member


HttpDispatcherAsync::HttpDispatcherAsync(boost::asio::io_context& io_context,
                                         HttpConnectionManage::ptr_s httpConnectionManage,
                                         HttpRequestMessage::ptr_s httpRequestMessage,
                                         HttpResponseMessage::ptr_s httpResponseMessage,
                                         int connect_timeout,
                                         int write_timeout,
                                         int read_timeout)
    : _io_context{io_context} ,
      _httpConnectionManage{httpConnectionManage} ,
      _httpRequestMessage{httpRequestMessage} ,
      _httpResponseMessage{httpResponseMessage},
      _connect_timeout(connect_timeout),
      _write_timeout(write_timeout),
      _read_timeout(read_timeout),
      _resolver(io_context),
      _deadline_timer(io_context),
      _httpResponseTranslation()
{

}

//设置 通知请求进度
void HttpDispatcherAsync::set_send_progress_callback(func_send_progress_callback&& send_progress_callback) {
    _send_progress_callback = std::move(send_progress_callback) ;
}
//设置 通知异步请求完成
void HttpDispatcherAsync::set_send_complete_callback(func_send_complete_callback&& send_complete_callback) {
    _send_complete_callback = std::move(send_complete_callback);
}

//执行异步请求
bool HttpDispatcherAsync::execute_request_async(){

    init();

    if(!_httpConnectionSharedPtr){
        _httpResponseMessage->setSendStatuCode(MP::HTTP::HTTP_SEND_STATU_CODE::ErrConnection);
        return false;
    }

    //如果连接已经打开，则直接发送数据，否则先进行连接
    if(_httpConnectionSharedPtr->isConnected()){
#if defined(MAPLE_MODE_DEBUGGING)
        MPLOG_INFO << ">>>>> Step1/2/3-> skip, use cache connection ";
#endif
        writeRequestStartUp();
    }
    else{
        resolveAsync(_httpRequestMessage->getHost(),_httpRequestMessage->getPort());
    }

    return true;
}

//初始化
void HttpDispatcherAsync::init(){

    //创建HTTP连接
    std::string port = std::to_string(_httpRequestMessage->getPort());
    _httpConnectionSharedPtr = _httpConnectionManage->getHttpConnection(_httpRequestMessage->getScheme(),
                                                                        _httpRequestMessage->getHost(),
                                                                        port,
                                                                        _httpRequestMessage->getSslSettings());


    _is_stopped = false;                               //标识请求是否已停止
    _is_deadline_timer_async_wait = false;             //标识定时器是否已调用async_wait
    _is_deadline_timer_enable = false;                 //标识定时器是否启用
    _write_payload_length = 0;                         //已写入数据大小

    //一次读取的服务端响应的数据容器
    if (_read_buf.size() != READ_RESPONSE_BUF_SIZE) {
        _read_buf.resize(READ_RESPONSE_BUF_SIZE);
    }

    _httpResponseTranslation.reset(_httpResponseMessage.get(),_httpRequestMessage->getHttpMethod());
}


//DNS解析
void HttpDispatcherAsync::resolveAsync(const std::string host,
                                       unsigned short port){

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << ">>>>> Step1-> resolve host:" << host << ":" << port;
#endif

    std::string strPort = boost::lexical_cast<std::string>(port);

    boost::asio::ip::tcp::resolver::query query(host,strPort);
    _resolver.async_resolve(query,
                            boost::bind(&HttpDispatcherAsync::resolveHandle, shared_from_this(),boost::asio::placeholders::error,boost::asio::placeholders::results));


}

void HttpDispatcherAsync::resolveHandle(const boost::system::error_code& err,
                                        const boost::asio::ip::tcp::resolver::results_type& endpoints){

    if(_is_stopped){
        return;
    }

    if(err){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrResolve,
                        err.message());
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    for (auto& endpoint : endpoints) {
        MPLOG_INFO << "host endpoint:" << endpointToString(endpoint);
    }
    MPLOG_INFO << "resolve host complete.";
#endif

    connectAsync(endpoints);
}

//建立Http连接
void HttpDispatcherAsync::connectAsync(const boost::asio::ip::tcp::resolver::results_type& endpoints){

    if(_is_stopped){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << ">>>>> Step2-> connect host endpoint";
#endif

    //设置连接超时定时器
    enableDeadlineTimer(_connect_timeout);
    //执行异步的连接
    boost::asio::async_connect(_httpConnectionSharedPtr->getHttpSocket(),
                               endpoints,
                               boost::bind(&HttpDispatcherAsync::connectHandle, shared_from_this(),boost::asio::placeholders::error));

}

void HttpDispatcherAsync::connectHandle(const boost::system::error_code& err){
    if(_is_stopped){
        return;
    }

    if(err){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrConnect,
                        err.message());
        return;
    }

    //修改状态（新增的）
    _httpConnectionSharedPtr->changeToConnect();

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "connect host complete.";
#endif

    //取消连接超时定时器
    disableDeadlineTimer();
    //发送连接已建立的事件回调
    progress_callback(1,0,0);

    //如果HTTPS，还需进行异步的握手校验
    if(_httpConnectionSharedPtr->isSsl()){
        handshakeAsync();
    }
    else{

#if defined(MAPLE_MODE_DEBUGGING)
        MPLOG_INFO << ">>>>> Step3-> it is not ssl , skip handshake .";
#endif
        writeRequestStartUp();
    }
}

//SSL握手
void HttpDispatcherAsync::handshakeAsync(){
    if(_is_stopped){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "Step3-> ssl socket handshake.";
#endif

    _httpConnectionSharedPtr->handshakeAsync(boost::bind(&HttpDispatcherAsync::handshakeHandle, shared_from_this(), _1));
}

void HttpDispatcherAsync::handshakeHandle(const boost::system::error_code& err){
    if(_is_stopped){
        return;
    }

    if(err){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrConnect,
                        err.message());
        //err != boost::asio::error::operation_aborted
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "ssl handshake complete.";
#endif

    writeRequestStartUp();
}

//发送数据
void HttpDispatcherAsync::writeRequestStartUp(){
    if(_is_stopped){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << ">>>>> Step4-> write data.";
#endif

    //设置连接超时定时器
    enableDeadlineTimer(_write_timeout);
    //开始写入Request Header
    writeRequestHeaderAsync();
}

void HttpDispatcherAsync::writeRequestHeaderAsync(){
    if(_is_stopped){
        return;
    }

    std::string strHeader;
    _httpRequestMessage->getRequestHeaderBuffers(strHeader);

    std::vector<boost::asio::const_buffer> buffers;
    buffers.push_back(boost::asio::buffer(strHeader));

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "write request header.\r\n" << strHeader;
#endif

    _httpConnectionSharedPtr->writeSomeAsync(buffers,
                                             boost::bind(&HttpDispatcherAsync::writeRequestHeaderHandle, shared_from_this(), _1, _2));


}

void HttpDispatcherAsync::writeRequestHeaderHandle(const boost::system::error_code& err,
                                                   std::size_t length){
    if(_is_stopped){
        return;
    }

    if(err){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrWrite,
                        err.message());
        //err != boost::asio::error::operation_aborted
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "write request header complete.";
#endif

    //取消连接超时定时器
    disableDeadlineTimer();

    if(_httpRequestMessage->getHttpRequestContent()->getContentLength() == 0){

#if defined(MAPLE_MODE_DEBUGGING)
        MPLOG_INFO << "payload content length = 0";
#endif
        //发送连接已建立的事件回调
        progress_callback(2,0,0);
        //开始接收Server端的响应数据
        readResponseDataStartUp();
    }
    else{
        writePayloadStartUp();
    }
}

void HttpDispatcherAsync::writePayloadStartUp(){
    if(_is_stopped){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "write request payload data.";
#endif

    if(!_httpRequestMessage->getHttpRequestContent()->beginWriteData()) {

        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrRequestData,
                        "http request content data error.");
        return;
    }

    writePayloadAsync();
}

void HttpDispatcherAsync::writePayloadAsync(){
    if(_is_stopped){
        return;
    }

    std::vector<boost::asio::const_buffer> buffers;
    //如果返回真，则表示仍有数据需要写入；如果返回假，则表示当前数据写入完成后则Payload写入完毕
    try {

        bool isContinue = _httpRequestMessage->getHttpRequestContent()->onWriteDataSome(buffers);

#if defined(MAPLE_MODE_DEBUGGING)
        //大于100KB则不在输出内容
        if(_httpRequestMessage->getHttpRequestContent()->getContentLength() < 102400){

            for(std::size_t i = 0;i< buffers.size();i++){
                boost::asio::const_buffer buf = buffers[i];
                std::size_t size = buf.size();
                const char* chars = static_cast<const char*>(buf.data());

                std::string tmp;
                tmp.append(chars,size);
                std::cout << SYS_TEXT(tmp);
            }
        }

#endif

        if(isContinue){
            _httpConnectionSharedPtr->writeSomeAsync(buffers,
                                                     boost::bind(&HttpDispatcherAsync::writePayloadHandle, shared_from_this(), _1, _2));
        }
        else{
            _httpConnectionSharedPtr->writeSomeAsync(buffers,
                                                     boost::bind(&HttpDispatcherAsync::writePayloadCompleteHandle, shared_from_this(), _1, _2));
        }



    } catch (...) {
        _httpRequestMessage->getHttpRequestContent()->endWriteData();

        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrWrite,
                        "write payload error." );
    }
}

void HttpDispatcherAsync::writePayloadHandle(const boost::system::error_code& err,
                                             std::size_t length){
    if(_is_stopped){
        return;
    }

    //处理数据写入异常
    if(err){
        _httpRequestMessage->getHttpRequestContent()->endWriteData();

        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrWrite,
                        err.message());
        //err != boost::asio::error::operation_aborted
        return;
    }


    //发送数据写入的事件回调
    _write_payload_length += length;

//#if defined(MAPLE_MODE_DEBUGGING)
//    MPLOG_INFO << "write length : " << _write_payload_length << "/" << _httpRequestMessage->getHttpRequestContent()->getContentLength();
//#endif

    progress_callback(2,_write_payload_length,_httpRequestMessage->getHttpRequestContent()->getContentLength());

    //继续写入
    writePayloadAsync();

}

void HttpDispatcherAsync::writePayloadCompleteHandle(const boost::system::error_code& err,
                                                     std::size_t length){

    if(_is_stopped){
        return;
    }

    _httpRequestMessage->getHttpRequestContent()->endWriteData();
    //处理数据写入异常
    if(err){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrWrite,
                        err.message() );
        //err != boost::asio::error::operation_aborted
        return;
    }

    //发送数据写入的事件回调
    _write_payload_length += length;
    progress_callback(2,_write_payload_length,_httpRequestMessage->getHttpRequestContent()->getContentLength());

    //最后需要检查一下写入的数据大小与HttpHeader中设置的大小（由程序计算）是否一致
    if(_write_payload_length != _httpRequestMessage->getHttpRequestContent()->getContentLength()){
        std::string err_msg =  "write data size not equal to HttpcontentLength. write_payload_length / HttpcontentLength :"
                + std::to_string(_write_payload_length) + " / " + std::to_string(_httpRequestMessage->getHttpRequestContent()->getContentLength());
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrWrite,err_msg);
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "write request payload data length : " << _write_payload_length << "/" << _httpRequestMessage->getHttpRequestContent()->getContentLength();
    MPLOG_INFO << "write request payload data completed.";
#endif

    readResponseDataStartUp();
}

//接收数据
void HttpDispatcherAsync::readResponseDataStartUp(){
    if(_is_stopped){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << ">>>>> Step5-> read response data.";
#endif

    //设置连接超时定时器
    enableDeadlineTimer(_read_timeout);

    //开始读取数据
    readResponseDataAsync();
}

void HttpDispatcherAsync::readResponseDataAsync(){
    if(_is_stopped){
        return;
    }

    _httpConnectionSharedPtr->readSomeAsync(boost::asio::buffer(_read_buf),
                                            boost::bind(&HttpDispatcherAsync::readResponseDataHandle, shared_from_this(), _1, _2));

}

void HttpDispatcherAsync::readResponseDataHandle(boost::system::error_code err,
                                                 std::size_t length){
    if(_is_stopped){
        return;
    }

    //取消连接超时定时器
    disableDeadlineTimer();

    //处理异常
    bool isEOF = (err == boost::asio::error::eof || err.value() == 0x140000db);
    if(err && !isEOF){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrRead,
                        "socket read aborted.");
        //err != boost::asio::error::operation_aborted
        return;
    }

    //解析数据
    if(!_httpResponseTranslation.parse(_read_buf.data(),length)){
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrRead,
                        "http response translation fail.");
        return;
    }

    //发送数据写入的事件回调
    if(_httpResponseTranslation.is_parsed_http_header()){
        progress_callback(3,_httpResponseTranslation.getContentReadLength(),_httpResponseTranslation.getContentLength());
    }

    //如果已经读取完毕
    if(_httpResponseTranslation.finished() || isEOF){
        readResponseDataComplete();
        return;
    }

    //继续读取
    readResponseDataAsync();
}

void HttpDispatcherAsync::readResponseDataComplete(){


    //数据读取完成
    _httpResponseMessage->getHttpResponseContent()->endReadData();

    //最后需要检查一下读取的数据大小与HttpHeader中设置的大小（程序解析）是否完整

    if(!_httpResponseTranslation.checkContentReadIsComplete()){
        std::string err_msg =  "read data size [ "
                + std::to_string(_httpResponseTranslation.getContentReadLength())
                + " ] not equal to contentLength [ "
                + std::to_string(_httpResponseTranslation.getContentLength()) + " ]";
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrRead,err_msg);
    }
    else{

#if defined(MAPLE_MODE_DEBUGGING)
        MPLOG_INFO << "";
        MPLOG_INFO << "read response data completed.";
#endif

        if(_httpResponseMessage->checkResponseCodeIsSuccess()){
            _httpResponseMessage->setSendStatuCode(HTTP_SEND_STATU_CODE::OK);
        }
        else{
            _httpResponseMessage->setSendStatuCode(HTTP_SEND_STATU_CODE::ErrNotHttp200);
        }

        end_request(true);
    }
}

//异常处理
void HttpDispatcherAsync::httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE err_code,
                                          const std::string& err_msg){

    _httpConnectionSharedPtr->cancle();

    _httpResponseMessage->setSendStatuCode(err_code);
    MPLOG_ERROR << "http error handle: " << HTTP_SEND_STATU_CODE_TOSTRING(err_code) << ","<< err_msg;

    end_request(false);
}

//结束请求
void HttpDispatcherAsync::end_request(bool isComplete){

    _is_stopped = true;
    _deadline_timer.cancel();
    _httpResponseMessage->getHttpResponseContent()->endReadData();

    //如果需要保存HTTP连接并且连接并未关闭，那么在pool中进行回收
    if(isComplete
            && _httpRequestMessage->getHttpRequestHeaders().getConnectionKeepalive()
            && _httpResponseMessage->getHttpResponseHeaders().getConnectionKeepalive()
            && _httpConnectionSharedPtr->isConnected()){
        _httpConnectionManage->recoveryHttpConnection(_httpConnectionSharedPtr);
    }

    //_httpConnectionSharedPtr.reset();     //不要再此次释放，否则SSL连接会报错，原因暂不明确

    //告知异步操作完成
    complete_callback(_httpResponseMessage);
}


//定时器相关函数
void HttpDispatcherAsync::enableDeadlineTimer(int seconds){

    if(seconds < 0){
        return;
    }

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "** enable deadline timer [ time limit " << seconds << " seconds ]";
#endif

    _deadline_timer.expires_after(std::chrono::seconds(seconds));
    _is_deadline_timer_enable = true;

    if(!_is_deadline_timer_async_wait){

        _deadline_timer.async_wait(
                    boost::bind(&HttpDispatcherAsync::deadlineTimerHandle, shared_from_this(), boost::asio::placeholders::error));

    }
    _is_deadline_timer_async_wait = true;
}

void HttpDispatcherAsync::disableDeadlineTimer(){

    if(!_is_deadline_timer_enable){
        return;
    }

    //将定时器的过期时间设置为最小值，以便在未修改定时器的过期时间之前，不再进入deadlineTimerHandle函数中
    _deadline_timer.expires_at(boost::asio::steady_timer::time_point::max());
    _is_deadline_timer_enable = false;

#if defined(MAPLE_MODE_DEBUGGING)
    MPLOG_INFO << "** disable deadline timer.";
#endif

}

void HttpDispatcherAsync::deadlineTimerHandle(const boost::system::error_code& err){

    if(_is_stopped){
        return;
    }

    //    if (err == boost::asio::error::operation_aborted) {
    //#if defined(MAPLE_MODE_DEBUGGING)
    //        MPLOG_INFO << "** deadline timer operation aborted ." << err.message();
    //#endif
    //    }

    //如果定时器处于启用状态 并且 定时器的失效时间 <= 当前时间，则判定为超时；否则继续等待
    if(_deadline_timer.expiry() <= boost::asio::steady_timer::clock_type::now()){

        //触发超时错误
        httpErrorHandle(MP::HTTP::HTTP_SEND_STATU_CODE::ErrTimeout,
                        "http operation timeout.");
        //将定时器的过期时间设置为最大值，以便在未修改定时器的过期时间之前，不再进入deadlineTimerHandle函数中
        //_deadline_timer.expires_at(steady_timer::time_point::max());
    }
    else{
        //#if defined(MAPLE_MODE_DEBUGGING)
        //        MPLOG_INFO << "** [Handle]deadline timer call async wait ." ;
        //#endif
        _deadline_timer.async_wait(
                    boost::bind(&HttpDispatcherAsync::deadlineTimerHandle, shared_from_this(), boost::asio::placeholders::error));
    }
}

//回调函数
void HttpDispatcherAsync::progress_callback(std::size_t step,std::size_t current,std::size_t total){
    if(_send_progress_callback){
        _send_progress_callback(step,current,total);
    }
}
void HttpDispatcherAsync::complete_callback(HttpResponseMessage::ptr_s httpResponseMessage){
    if(_send_complete_callback){
        _send_complete_callback(httpResponseMessage);
    }
}


/*****************************************
 * HttpDispatcher
 *****************************************/

//执行同步请求
void HttpDispatcher::execute_request(){

    //首先对互斥量_async_request_mutex加锁，并将变量_async_request_finished设置成false,并启动异步执行逻辑
    std::unique_lock<std::mutex> lk(_async_request_mutex);
    _async_request_finished = false;
    if(execute_request_async()){
        //调用_async_request_cv的wait函数对互斥量进行解锁，阻塞当前线程，等待 execute_request_async 完成时将_async_request_finished重置为true
        _async_request_cv.wait(lk, [=]{ return _async_request_finished;});
    }
}

void HttpDispatcher::end_request(bool isComplete){

    HttpDispatcherAsync::end_request(isComplete);

    //如果有同步线程处于等待状态，则执行execute_request的阻塞唤醒逻辑
    _async_request_mutex.lock();
    if (!_async_request_finished) {
        _async_request_finished = true;
        _async_request_mutex.unlock();
        _async_request_cv.notify_one();
    } else {
        _async_request_mutex.unlock();
    }
}


}   //INTERNAL
}   //HTTP
}   //MP
